package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.DimJoinFunction;
import com.atguigu.edu.realtime.util.DimUtil;
import com.atguigu.edu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @Author zhangsan
 * @Date 2022/10/19 18:53
 * @Description //TODO 发送异步请求完成维度关联
 * 模板方法设计模式
 *        在父类中定义完成某一个功能的核心算法的骨架，具体的实现延迟到子类中去完成。
 *       在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 */
public abstract class DimAsyncFunction <T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {
    private ExecutorService executorService;

    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //开启多个线程 发送异步请求  完成维度关联
        executorService.submit(
                new Runnable() {
                    @Override
                    public void run() {
                        //根据流中的对象获取要关联的维度主键
                        String key = getKey(obj);
                        //System.out.println("key" + key);
                        //根据维度的主键获取维度对象
                        JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                        System.out.println("dimInfoJsonObj"+dimInfoJsonObj);
                        //将维度对象的属性补充到流中的对象
                        if(dimInfoJsonObj != null){
                            join(obj,dimInfoJsonObj);
                        }
                        //将关联之后的对象向下游传递
                        resultFuture.complete(Collections.singleton(obj));
                    }
                }
        );
    }



}
